package com.mtw.bbs.message.mq;



import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mtw.bbs.common.core.constant.MessageConstant;
import com.mtw.bbs.message.pojo.entity.Message;
import com.mtw.bbs.message.service.MessageService;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeanUtils;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;

/**
 * 文章消息消费者
 */
@Component
@RequiredArgsConstructor
public class KafkaMessageConsumer {


    private final MessageService messageService;


    /**
     * 消费评论点赞
     */
    @KafkaListener(topics = {"commentLike"})
    public void commentLikeSaveMessage(List<ConsumerRecord<String, String>> record) {

        if (!CollectionUtils.isEmpty(record)) {

            JSONObject jsonObject = JSON.parseObject(record.get(0).value());
            Map<String,Object> map = jsonObject;

            Message message = new Message();
            message.setType(MessageConstant.MSG_LIKE_COMMENT);
            message.setBillId((String) map.get("billId"));
            message.setReceiver((String) map.get("receiver"));
            message.setCreatedBy((String) map.get("createdBy"));

            messageService.saveMessage(message);
        }

    }


    /**
     * 消费评论
     */
    @KafkaListener(topics = {"commentMessage"})
    public void commentMessage(List<ConsumerRecord<String, String>> record) {
        if (!CollectionUtils.isEmpty(record)) {

            JSONObject jsonObject = JSON.parseObject(record.get(0).value());
            Map<String,Object> map = jsonObject;

            Message message = new Message();
            message.setType((Integer) map.get("type"));
            message.setBillId((String) map.get("billId"));
            message.setBillParentId((String) map.get("billParentId"));
            message.setReceiver((String) map.get("receiver"));
            message.setCreatedBy((String) map.get("createdBy"));

            messageService.saveMessage(message);
        }

    }


    /**
     * 消费文章点赞
     */
    @KafkaListener(topics = {"articleLike"})
    public void articleLikeMessage(List<ConsumerRecord<String, String>> record) {

        if (!CollectionUtils.isEmpty(record)) {

            JSONObject jsonObject = JSON.parseObject(record.get(0).value());
            Map<String,Object> map = jsonObject;

            Message message = new Message();
            message.setType(MessageConstant.MSG_LIKE_ARTICLE);
            message.setBillId((String) map.get("billId"));
            message.setReceiver((String) map.get("receiver"));
            message.setCreatedBy((String) map.get("createdBy"));

            messageService.saveMessage(message);
        }

    }
}
